package remoting.grpc;

import com.google.protobuf.ByteString;
import common.exception.RemotingException;
import common.model.Codec2;
import common.model.MagicUrl;
import common.model.Request;
import common.model.RpcInvocation;
import common.util.NetUtils;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import remoting.channel.Channel;
import remoting.channel.ChannelHandler;
import remoting.channel.buffer.ChannelBuffer;
import remoting.channel.buffer.ChannelBuffers;
import remoting.channel.buffer.DynamicChannelBuffer;
import remoting.grpc.proto.GrpcReply;
import remoting.grpc.proto.GrpcRequest;

public class GrpcHandler extends GreeterGrpc.GreeterImplBase {

  private static final Logger logger = LoggerFactory.getLogger(GrpcHandler.class);

  private final Map<String, Channel> channels = new ConcurrentHashMap<String, Channel>();

  private final MagicUrl url;

  private final ChannelHandler handler;

  private final Codec2 codec;

  private ChannelBuffer previousData = ChannelBuffers.EMPTY_BUFFER;

  public GrpcHandler(MagicUrl url, ChannelHandler handler, Codec2 codec) {
    this.url = url;
    this.handler = handler;
    this.codec = codec;
  }

  @Override
  public StreamObserver<GrpcRequest> getRp(StreamObserver<GrpcReply> responseObserver) {
    StreamObserver<GrpcRequest> streamObserver = new StreamObserver<GrpcRequest>() {
      GrpcServerChannel channel = GrpcServerChannel.getOrAddChannel(responseObserver, url, handler, codec);

      @Override
      public void onNext(GrpcRequest grpcRequest) {

        channel.setLocalAddress(new InetSocketAddress(url.getPort()));
        ByteString grpcRequestData = grpcRequest.getData();
        try {
          byte[] bytes = grpcRequestData.toByteArray();
          int bufferSize = bytes.length;
          ChannelBuffer frame;

          if (previousData.readable()) {
            if (previousData instanceof DynamicChannelBuffer) {
              previousData.writeBytes(bytes);
              frame = previousData;
            } else {
              int size = previousData.readableBytes() + bufferSize;
              frame = ChannelBuffers.dynamicBuffer(size > bufferSize ? size : bufferSize);
              frame.writeBytes(previousData, previousData.readableBytes());
              frame.writeBytes(bytes);
            }
          } else {
            frame = ChannelBuffers.wrappedBuffer(bytes);
          }
          Object object = codec.decode(channel, frame);
          logger.debug("grpc server read {}", object);
          if(object instanceof Request){
            Request request = (Request) object;
            Object data = request.getData();
            if(data instanceof RpcInvocation){
              RpcInvocation rpcInvocation = (RpcInvocation) data;
              String addr = rpcInvocation.getHost();
              Integer port = rpcInvocation.getPort();
              InetSocketAddress inetSocketAddress = new InetSocketAddress(addr, port);
              channel.setRemoteAddress(inetSocketAddress);
              if (channel != null) {
                channels.put(NetUtils.toAddressString(inetSocketAddress), channel);
              }
            }
          }
          handler.received(channel, object);
        } catch (RemotingException remotingException) {
          logger.error("Grpc onNext error {}", remotingException);
        } catch (IOException ioe){
          logger.error("Grpc codec.decode error {}", ioe);
        }
      }

      @Override
      public void onError(Throwable throwable) {
        try {
          handler.disconnected(channel);
        } catch (RemotingException remotingException) {
          remotingException.printStackTrace();
        } finally {
          GrpcServerChannel.removeChannelIfDisconnected(responseObserver);
        }
      }

      @Override
      public void onCompleted() {
        //responseObserver.onCompleted();
      }
    };
    return streamObserver;
  }

  public Map<String, Channel> getChannels() {
    return channels;
  }
}
